/**
 * 
 */
package component.activemq;

import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 
 * @author liuboen
 * 
 * since:2012-6-7
 * version:1.0.0
 */
public class TopicPersistentReceive {

	/**
	 * @param args
	 * @throws JMSException 
	 */
	public static void main(String[] args) throws JMSException {
		// ConnectionFactory ：连接工厂，JMS 用它创建连接 
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory( 
                        ActiveMQConnection.DEFAULT_USER, 
                        ActiveMQConnection.DEFAULT_PASSWORD, 
                        "tcp://127.0.0.1:61616"); 
		
		TopicConnection topicConnection = connectionFactory.createTopicConnection();  
		//设置订阅者的ID，以后再用这个身份的话再连接的话，会将此ID没有接收到的消息接收下来（也就是离线消息）  
		topicConnection.setClientID("test-topic-client");  
		final TopicSession session = topicConnection.createTopicSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
		
		Topic topic = session.createTopic("publish-topic");
		//创建持久化订阅对象  
		TopicSubscriber subscriber = session.createDurableSubscriber(topic, "Subscriber_name_test");
		//消费者消息监听
		subscriber.setMessageListener(new MessageListener() {
			public void onMessage(Message message) {

				MapMessage mapMessage = (MapMessage) message;
				if (null != mapMessage) {
					try {
						System.out.println("收到消息：" + mapMessage.getString("tst"));
						session.commit();
					} catch (JMSException e) {
						e.printStackTrace();
					}
				}

			}

		});
		topicConnection.start();
         // session.commit();
		
          
	}

}
